package com.world.rickcloudendpoint.service.net.websocket;

import com.alibaba.fastjson.JSON;
import com.world.rickcloudendpoint.common.constant.GlobalConstants;
import com.world.rickcloudendpoint.common.enums.Cmd;
import io.netty.channel.Channel;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.util.AttributeKey;
import io.netty.util.internal.StringUtil;
import lombok.extern.log4j.Log4j2;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @description: web socket topic的订阅与取消
 * @author: jlsong
 * @date: 2021/12/12 21:23:42
 * @version: 1.0
 */
@Component
@Log4j2
public class TopicSubscribeManager {

    /**
     * 缓存通道连接订阅的主题
     */
    private static final ConcurrentHashMap<String, HashMap<String, SessionInfo>> CACHE = new ConcurrentHashMap<>();

    private static final AttributeKey<String> CHANNEL_ATTR_TOPIC_KEY = AttributeKey.valueOf(GlobalConstants.CHANNEL_TOPIC_ATTR_KEY);
    private static final AttributeKey<String> CHANNEL_ATTR_USER_KEY = AttributeKey.valueOf(GlobalConstants.CHANNEL_USER_ATTR_KEY);

    public static final String topicSplitKey = GlobalConstants.WEB_SOCKET_SUBSCRIBE_TOPIC_PATH.concat("/");
    public static final String uidSplitKey = GlobalConstants.WEB_SOCKET_SUBSCRIBE_UID_PATH.concat("/");


    /**
     * 主题订阅
     *
     * @param topic   主题
     * @param uid     用户
     * @param channel 连接通道
     */
    public void register(String topic, String uid, Channel channel) {
        if (StringUtil.isNullOrEmpty(topic) || StringUtil.isNullOrEmpty(uid) || channel == null) {
            log.debug("{}订阅的主题或用户或连接通道值存在空值.....", GlobalConstants.WEB_SOCKET_LOG_ID);
            return;
        }
        channel.attr(CHANNEL_ATTR_TOPIC_KEY).set(topic);
        channel.attr(CHANNEL_ATTR_USER_KEY).set(uid);
        HashMap<String, SessionInfo> map = CACHE.get(topic);
        if (map == null) {
            map = new HashMap<>(16);
        }
        SessionInfo session = new SessionInfo();
        session.setSid(channel.id().asLongText());
        session.setTopic(topic);
        session.setUid(uid);
        session.setChannel(channel);
        map.put(session.getSid(), session);
        CACHE.put(topic, map);
    }


    /**
     * 取消订阅主题
     *
     * @param channel 连接通道
     */
    public void unRegister(Channel channel) {
        if (channel == null) {
            log.debug("{}取消订阅的连接不存在.....", GlobalConstants.WEB_SOCKET_LOG_ID);
            return;
        }
        String topic = getTopic(channel);
        if (StringUtil.isNullOrEmpty(topic)) {
            return;
        }
        HashMap<String, SessionInfo> map = CACHE.get(topic);
        if (map == null) {
            map = new HashMap<>(16);
            CACHE.put(topic, map);
            return;
        }
        String uid = getUid(channel);
        if (StringUtil.isNullOrEmpty(uid)) {
            return;
        }
        map.remove(channel.id().asShortText());
        CACHE.put(topic, map);
    }

    public String getTopic(Channel channel) {
        return channel.attr(CHANNEL_ATTR_TOPIC_KEY).get();
    }

    public String getUid(Channel channel) {
        return channel.attr(CHANNEL_ATTR_USER_KEY).get();
    }


    /**
     * 通过路由，截取主题和用户id
     *
     * @param requestUri 路由
     * @param channel    连接通道
     * @return 用户订阅主题信息
     */
    public SessionInfo getSessionFromRequestRui(String requestUri, Channel channel) {
        int endIndex = requestUri.indexOf("?");
        String topicKey = "";
        String uidKey = "";
        if (requestUri.indexOf(topicSplitKey) == -1 || requestUri.indexOf(uidSplitKey) == -1){
            return null;
        }
        if (endIndex != -1) {
            topicKey = requestUri.substring(requestUri.indexOf(topicSplitKey) + topicSplitKey.length(), requestUri.indexOf("?"));
            uidKey = requestUri.substring(requestUri.indexOf(uidSplitKey) + uidSplitKey.length(), requestUri.indexOf("?"));
        } else {
            topicKey = requestUri.substring(requestUri.indexOf(topicSplitKey) + topicSplitKey.length());
            uidKey = requestUri.substring(requestUri.indexOf(uidSplitKey) + uidSplitKey.length());
        }
        if (!StringUtil.isNullOrEmpty(topicKey) && !StringUtil.isNullOrEmpty(uidKey)) {
            int topicValueSplit = topicKey.indexOf("/");
            String topic = "";
            if (topicValueSplit != -1) {
                topic = topicKey.substring(0, topicValueSplit);
            } else {
                topic = topicKey;
            }
            if (topic != null && topic.trim().length() > 0) {
                // 检验用户uid
                int uidValueSplit = uidKey.indexOf("/");
                String uid = "";
                if (uidValueSplit != -1) {
                    uid = uidKey.substring(0, uidValueSplit);
                } else {
                    uid = uidKey;
                }
                log.info("=================> topic={}  ;  uid={}  ", topic, uid);
                if (!StringUtil.isNullOrEmpty(topic) && !StringUtil.isNullOrEmpty(uid)) {
                    SessionInfo sessionInfo = new SessionInfo();
                    sessionInfo.setSid(channel.id().asLongText());
                    sessionInfo.setTopic(topic);
                    sessionInfo.setUid(uid);
                    sessionInfo.setChannel(channel);
                    return sessionInfo;
                }
            }
        }
        return null;
    }


    /**
     * 给某个用户关联的通道发送消息
     *
     * @param cmd   消息命令
     * @param topic 订阅的主题
     * @param obj   数据
     */
    public void send(Cmd cmd, String topic, Object obj) {
        HashMap<String, SessionInfo> sessions = CACHE.get(topic);
        if (sessions == null) {
            return;
        }

        Map<String, Object> data = new LinkedHashMap<>();
        data.put("cmd", cmd.getCmd());
        data.put("data", obj);
        String msg = JSON.toJSONString(data);
        log.info(String.format("%s 服务器下发消息: %s", GlobalConstants.WEB_SOCKET_LOG_ID, msg));
        sessions.values().forEach(session -> {
            session.getChannel().writeAndFlush(new TextWebSocketFrame(msg));
        });
    }

    /**
     * 给某个通道发送消息
     *
     * @param channel 连接通道
     * @param cmd     消息命令
     * @param obj     数据
     */
    public void send(Channel channel, Cmd cmd, Object obj) {
        Map<String, Object> data = new LinkedHashMap<>();
        data.put("cmd", cmd.getCmd());
        data.put("data", obj);
        String msg = JSON.toJSONString(data);
        log.info(String.format("%s 服务器下发消息: %s", GlobalConstants.WEB_SOCKET_LOG_ID, msg));
        channel.writeAndFlush(new TextWebSocketFrame(msg));
    }
}
